package pool

import (
	"errors"
	"fmt"
	"log"
	"sync"
	"sync/atomic"
	"time"
)

// Pool 连接池接口
type Pool interface {
	// Init 初始化连接池
	// Init()
	// Get 获取连接池中的连接
	Get() (Conn, error)
	// Put 向连接池中放连接
	Put(Conn) error
	// New 新建一个空闲连接
	New() error
	// Shut 关闭第一个空闲连接（理论上它就是存在最长的）		可改进为关闭一个持续最久的连接
	Shut(Conn) error
	// Close 关闭连接池，释放连接
	Close() error
	// Status 连接池状态
	Status() string
	// Scan 周期性扫描连接信息			检查各类连接数量是否符合要求，进行新建or关闭；检查空闲连接是否超时，超时则关闭
	Scan() error
}

type pool struct {
	// 原子变量，池内当前空闲物理连接数量
	current int32
	// 原子变量, 已占用物理连接
	working int32
	// pool options
	opt Options
	// 所有空闲的物理连接
	conns []*conn
	// 服务端地址
	address string
	// closed Close函数引用时为true
	closed int32
	// called 单位时间建立一次性连接的个数，过多则停止新建，转而让新请求等待
	called int32
	// busy 表示当前池连接过多，此时停止新建，转而让新请求等待
	busy bool
	// channel 表示等待队列
	channel chan *conn

	// 控制变量读写
	sync.RWMutex
}

// 初始化 grpc 返回一个连接池.
func NewPool(address string, op *Options) (Pool, error) {
	option := DefaultOptions
	option.Dial = DialTest
	if op != nil {
		option = *op
	}

	// 检查输入和option的问题
	if address == "" {
		return nil, errors.New("invalid address settings")
	}
	if option.Dial == nil {
		return nil, errors.New("invalid dial settings")
	}
	if option.MaxIdle <= 0 || option.MaxActive <= 0 || option.MaxIdle > option.MaxActive || option.MaxIdle < option.MinIdle {
		return nil, errors.New("invalid maximum settings")
	}

	// 结构体初始化
	p := &pool{
		//index:   0,
		current: 0,
		working: 0,
		opt:     option,
		conns:   make([]*conn, option.MaxActive),
		//status:  make([]int32, option.MaxActive+option.MaxIdle),
		address: address,
		closed:  0,
	}

	// 为每个空闲创建连接
	for i := 0; i < p.opt.MinIdle; i++ {
		err := p.New()
		if err != nil {
			p.Close()
			return nil, fmt.Errorf("dial is not able to fill the pool: %s", err)
		}
	}
	log.Printf("new pool success: %v\n", p.Status())

	return p, nil
}

func (p *pool) New() error {
	c, err := p.opt.Dial(p.address)
	index := atomic.LoadInt32(&p.current)
	if err != nil {
		return fmt.Errorf("%s", err)
	}
	p.Lock()
	p.conns[index] = p.wrapConn(c, false)
	atomic.AddInt32(&p.current, 1)
	p.Unlock()
	return nil
}

func (p *pool) Get() (Conn, error) {
	// 从已建立的连接中选择一个
	p.RLock()
	current := atomic.LoadInt32(&p.current)
	working := atomic.LoadInt32(&p.working)
	p.RUnlock()

	// 如果有空闲连接，丢一个出去(这里是丢第一个给用户，跟队列似的)
	if current != 0 {
		p.Lock()
		connect := p.conns[0]
		p.conns = p.conns[1:]
		p.conns = append(p.conns, nil) // 为了维持原数组数量不变
		atomic.AddInt32(&p.working, 1)
		atomic.AddInt32(&p.current, -1)
		p.Unlock()
		return connect, nil
	}

	// 下面是没有空闲连接的情况，根据情况进行扩容操作

	// 如果连接数达到最大活跃连接数，则单独建立grpc普通链接			这里可以改进：如果请求过多，则让新请求等待
	if working >= int32(p.opt.MaxActive) {
		// 如果没那么忙，建立一次性连接
		atomic.AddInt32(&p.called, 1)
		if p.busy == false {
			c, err := p.opt.Dial(p.address)
			return p.wrapConn(c, true), err
		}
		// 建立的临时连接过多，剩下的请求要求等会（接受从put直接传来的conn
		if p.called >= 10 {
			connect := <-p.channel
			return connect, nil
		}
	}

	// 如果没达到最大活跃连接数，按照特定的增长策略批量创建新连接: 创建MI个
	p.Lock()
	increment := int32(p.opt.MinIdle)
	if increment > int32(p.opt.MaxActive)-working {
		increment = int32(p.opt.MaxActive) - working
	}

	for i := 0; i < int(increment); i++ {
		_ = p.New()
	}

	log.Printf("added %d connects by Get()", increment)
	atomic.AddInt32(&p.working, 1)
	atomic.AddInt32(&p.current, -1)
	p.Unlock()
	return p.conns[0], nil
}

func (p *pool) Put(c Conn) error {
	atomic.AddInt32(&p.working, -1)
	current := atomic.LoadInt32(&p.current)
	working := atomic.LoadInt32(&p.working)

	// 几种丢弃的情况
	// 如果c是一次性连接，则丢弃
	if c.Once() {
		c.Close()
		return nil
	}

	// 如果busy,没有空闲连接，有用户在等待连接，则归还后直接移交连接
	if p.busy == true {
		p.channel <- p.wrapConn(c.Value(), false)
		return nil
	}

	// 如果池满，则丢弃
	if current+working >= int32(p.opt.MaxActive) {
		c.Close()
		return nil
	}

	// 下面是有空余位置可以归还连接的情况
	p.Lock()
	p.conns[current] = p.wrapConn(c.Value(), false)
	atomic.AddInt32(&p.current, 1)
	current = atomic.LoadInt32(&p.current)
	p.Unlock()

	return nil
}

func (p *pool) Shut(c Conn) error {
	if c == nil {
		p.Lock()
		p.conns[0].Close()
		p.conns = p.conns[1:]
		p.conns = append(p.conns, nil) // 为了维持原数组数量不变
		atomic.AddInt32(&p.current, -1)
		p.Unlock()
		return nil
	}
	panic("other input")
}

// Status 查看连接池状态
func (p *pool) Status() string {
	return fmt.Sprintf("address:%s, option:%v, idle_conn:%d, active_conn:%d,",
		p.address, p.opt, p.current, p.working)
}

func (p *pool) Scan() error {
	for {
		time.Sleep(200 * time.Millisecond)

		// 如果两次扫描间临时建立的连接超过10个，进入busy状态，如果没有临时连接就是空闲
		if p.called >= 10 {
			p.busy = true
		}
		if p.called == 0 {
			p.busy = false
		}

		// 计数器归零，每次扫描归零一次，这样它表示的就是单位时间的临时连接建立数
		atomic.StoreInt32(&p.called, 0)

		for i := 0; i < p.opt.MaxActive; i++ {
			if p.conns[i] != nil {
				p.conns[i].AddTime()
			}
		}
		for {
			if p.conns[0].time == 5 {
				p.Shut(nil)
			}
			if p.conns[0].time < 5 {
				break
			}
		}

		p.adjustCapacity()
	}
}

// Close 关闭连接池
func (p *pool) Close() error {
	atomic.StoreInt32(&p.closed, 1)
	atomic.StoreInt32(&p.current, 0)
	atomic.StoreInt32(&p.working, 0)
	p.deleteFrom(0)
	log.Printf("close pool success: %v\n", p.Status())
	return nil
}

// 下面这些是子函数
func (p *pool) close(index int) {
	conn := p.conns[index]
	if conn == nil {
		return
	}
	conn.Close()
	p.conns[index] = nil
}

func (p *pool) deleteFrom(begin int) {
	for i := begin; i < p.opt.MaxActive; i++ {
		p.close(i)
	}
}

// 读取current和working并进行扩缩容
func (p *pool) adjustCapacity() {
	current := atomic.LoadInt32(&p.current)
	working := atomic.LoadInt32(&p.working)
	if current > int32(p.opt.MaxIdle) {
		minus := current - int32(p.opt.MaxIdle)
		for i := 0; i < int(minus); i++ {
			p.Shut(nil)
		}
	}
	if current < int32(p.opt.MinIdle) {
		add := int32(p.opt.MinIdle) - current
		if add > int32(p.opt.MaxActive)-current-working {
			add = int32(p.opt.MaxActive) - current - working
		}
		for i := 0; i < int(add); i++ {
			p.New()
		}
	}
}

//func (p *pool) incrRef() int32 {
//	newRef := atomic.AddInt32(&p.ref, 1)
//	if newRef == math.MaxInt32 {
//		panic(fmt.Sprintf("overflow ref: %d", newRef))
//	}
//	return newRef
//}
//
//func (p *pool) decrRef() {
//	newRef := atomic.AddInt32(&p.ref, -1)
//	if newRef < 0 && atomic.LoadInt32(&p.closed) == 0 {
//		panic(fmt.Sprintf("negative ref: %d", newRef))
//	}
//	if newRef == 0 && atomic.LoadInt32(&p.current) > int32(p.opt.MaxIdle) {
//		p.Lock()
//		if atomic.LoadInt32(&p.ref) == 0 {
//			log.Printf("shrink pool: %d ---> %d, decrement: %d, maxActive: %d\n",
//				p.current, p.opt.MaxIdle, p.current-int32(p.opt.MaxIdle), p.opt.MaxActive)
//			atomic.StoreInt32(&p.current, int32(p.opt.MaxIdle))
//			p.deleteFrom(p.opt.MaxIdle)
//		}
//		p.Unlock()
//	}
//}
